package com.lingjtx.common.thread.core;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.function.Supplier;
import java.util.stream.Collectors;

@Component
public class CompletableFutureTask {

    private static final Logger log = LoggerFactory.getLogger(CompletableFutureTask.class);

    private final ThreadPoolManager threadPoolManager;

    private String poolName = PoolNameConst.POOL_DEFAULT;

    public CompletableFutureTask(ThreadPoolManager threadPoolManager) {
        this.threadPoolManager = threadPoolManager;
    }

    public String getPoolName() {
        return poolName;
    }

    public void setPoolName(String poolName) {
        this.poolName = poolName;
    }

    // ========== 基础执行 ==========

    // 使用默认线程池执行任务
    public CompletableFuture<Void> runAsync(Runnable runnable) {
        return runAsync(runnable, poolName); // 使用 "default" 线程池
    }

    public CompletableFuture<Void> runAsync(Runnable runnable, String poolName) {
        ExecutorService executor = getExecutor(poolName);
        return CompletableFuture.runAsync(runnable, executor);
    }

    // 使用指定的线程池执行任务
    public CompletableFuture<Void> runAsync(Runnable runnable, String poolName, ExceptionCallback callback) {
        ExecutorService executor = getExecutor(poolName);
        return CompletableFuture.runAsync(wrap(runnable, callback), executor);
    }

    public <T> CompletableFuture<T> supplyAsync(Supplier<T> supplier) {
        return supplyAsync(supplier, poolName);
    }

    public <T> CompletableFuture<T> supplyAsync(Supplier<T> supplier, String poolName) {
        ExecutorService executor = getExecutor(poolName);
        return CompletableFuture.supplyAsync(supplier, executor);
    }

    public <T> CompletableFuture<T> supplyAsync(Supplier<T> supplier, String poolName, ExceptionCallback callback) {
        ExecutorService executor = getExecutor(poolName);
        return CompletableFuture.supplyAsync(wrap(supplier, callback), executor);
    }


    // ========== 多个一样的任务转换 ==========
    public <T, R> List<CompletableFuture<R>> convert(T[] arr, TaskCallback<T, R> callback) {
        return Arrays.stream(arr)
                .map(item -> supplyAsync(() -> callback.handle(item)))
                .toList();
    }

    // 支持任意 Collection (List, Set)
    public <T, R> List<CompletableFuture<R>> convert(Collection<T> collection, TaskCallback<T, R> callback) {
        return collection.stream()
                .map(item -> supplyAsync(() -> callback.handle(item)))
                .toList();
    }

    // 支持 Map<K, V>，默认传 Entry<K,V> 给 callback
    public <K, V, R> List<CompletableFuture<R>> convert(Map<K, V> map, TaskCallback<Map.Entry<K, V>, R> callback) {
        return map.entrySet().stream()
                .map(entry ->supplyAsync(() -> callback.handle(entry)))
                .toList();
    }

    // 支持 varargs
    @SafeVarargs
    public final <T, R> List<CompletableFuture<R>> convert(TaskCallback<T, R> callback, T... obs) {
        return Arrays.stream(obs)
                .map(item -> supplyAsync(() -> callback.handle(item)))
                .toList();
    }

    // ========== List 聚合任务 ==========

    // 等待所有任务完成
    public CompletableFuture<Void> allOf(List<CompletableFuture<?>> futures) {
        return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
    }

    // 返回第一个完成的任务
    public CompletableFuture<Object> anyOf(List<CompletableFuture<?>> futures) {
        return CompletableFuture.anyOf(futures.toArray(new CompletableFuture[0]));
    }

    public <T> List<T> joinAll(List<CompletableFuture<T>> futures) {
        return futures.stream()
                .map(CompletableFuture::join)
                .collect(Collectors.toList());
    }

    // 异步方式收集所有任务结果
    public CompletableFuture<List<?>> sequence(List<CompletableFuture<?>> futures) {
        return allOf(futures).thenApply(v ->
                futures.stream().map(CompletableFuture::join).collect(Collectors.toList()));
    }

    // ========== Map 任务处理 ==========

    // 等待所有 map 的任务完成（不返回结果）
    public CompletableFuture<Void> allOfMap(Map<String, CompletableFuture<?>> futures) {
        return CompletableFuture.allOf(futures.values().toArray(new CompletableFuture[0]));
    }

    // 返回第一个完成的任务及其 key（anyOf）
    public CompletableFuture<Map.Entry<String, Object>> anyOfMap(Map<String, CompletableFuture<?>> futures) {
        CompletableFuture<Object> any = CompletableFuture.anyOf(futures.values().toArray(new CompletableFuture[0]));

        return any.thenApply(result -> {
            for (Map.Entry<String, CompletableFuture<?>> entry : futures.entrySet()) {
                if (entry.getValue().isDone()) {
                    try {
                        Object value = entry.getValue().getNow(null); // 避免阻塞
                        if (Objects.equals(value, result)) {
                            return Map.entry(entry.getKey(), result);
                        }
                    } catch (Exception ignored) {
                    }
                }
            }
            return null;
        });
    }

    // join 形式阻塞收集 map 结果
    public <T> Map<String, T> joinAllMap(Map<String, CompletableFuture<T>> futures) {
        Map<String, T> result = new HashMap<>();
        futures.forEach((key, future) -> {
            try {
                result.put(key, future.join());
            } catch (Exception e) {
                result.put(key, null); // 异常处理方式可改为抛出或日志
            }
        });
        return result;
    }

    // 异步 sequence 收集结果
    public <T> CompletableFuture<Map<String, T>> sequenceMap(Map<String, CompletableFuture<T>> futures) {
        return CompletableFuture.allOf(futures.values().toArray(new CompletableFuture[0]))
                .thenApply(v -> {
                    Map<String, T> result = new ConcurrentHashMap<>();
                    futures.forEach((key, future) -> {
                        try {
                            result.put(key, future.join());
                        } catch (Exception e) {
                            result.put(key, null);
                        }
                    });
                    return result;
                });
    }

    // ========== 工具方法 ==========

    private ExecutorService getExecutor(String poolName) {
        ExecutorService executor = threadPoolManager.get(poolName);
        return executor != null ? executor : threadPoolManager.getOrCreate(this.poolName);
    }

    private Runnable wrap(Runnable task, ExceptionCallback callback) {
        return () -> {
            try {
                task.run();
            } catch (Exception e) {
                if (callback != null) {
                    callback.handle(e);
                }
            }
        };
    }

    private <T> Supplier<T> wrap(Supplier<T> supplier, ExceptionCallback callback) {
        return () -> {
            try {
                return supplier.get();
            } catch (Exception e) {
                if (callback != null) {
                    callback.handle(e);
                }
                return null;
            }
        };
    }

    public void shutdown(String poolName) {
        ExecutorService executor = threadPoolManager.get(poolName);
        if (executor != null) {
            executor.shutdown();
        }
    }

    public void shutdownNow(String poolName) {
        ExecutorService executor = threadPoolManager.get(poolName);
        if (executor != null) {
            executor.shutdownNow();
        }
    }

    public void shutdownAll() {
        threadPoolManager.getAllPools().forEach((k, v) -> v.shutdown());
    }

    @FunctionalInterface
    public interface ExceptionCallback {
        void handle(Exception ex);
    }

    @FunctionalInterface
    public interface TaskCallback<T, R> {
        R handle(T t);
    }
}